package test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object UserVisiterAction_Pageflow {
  def main(args: Array[String]): Unit = {
    val spark = new SparkConf().setMaster("local[*]").setAppName("UserVisitActionTest04")
    val sc = new SparkContext(spark)
    sc.setLogLevel("ERROR")
    //读取文件
    val sourceRDD = sc.textFile("data/user_visit_action.txt")
    val actionDataRDD = sourceRDD.map(
      action => {
        val datas = action.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    //TODO 计算分母
    val pageIDToCountMap = actionDataRDD.map(
      action => {
        (action.page_id, 1l)
      }
    ).reduceByKey(_ + _).collect().toMap

    //todo 计算分子
    //根据session来进行分组
    val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)
    //根据时间来进行排序
    val mvRDD = sessionRDD.mapValues(
      iter => {
        val sortList = iter.toList.sortBy(_.action_time)
        //求出按照时间顺序排序的连续的访问网页的的一个顺序
        /**
         * 此时page_id也就是网页的访问顺序如下
         * [1,2,3,4,5]
         * 但是我们需要其变成
         * [1-2,2-3,3-4,4-5]
         * 可以使用zip（拉链）[1,2,3]zip[2,3,4](拉链就是第一个rdd做key，第二个相同位置的rdd做value)
         */
        val flowIds: List[Long] = sortList.map(_.page_id)
        //TODO tail方法是取除第一个之外的全部元素（列表里的）
        val pageflowIds: List[(Long, Long)] = flowIds.zip(flowIds.tail)
        pageflowIds.map((_, 1))
      }
    )
    val flatRDD = mvRDD.map(_._2).flatMap(list => list)

    val dataRDD = flatRDD.reduceByKey(_+_)

    //TODO 计算单跳转换率（分子/分母）
    dataRDD.foreach{
      case ((pageId1,pageId2),sum) =>{
        val pageCount = pageIDToCountMap.getOrElse(pageId1,0l)
        println(s"页面${pageId1}跳转到页面${pageId2}单跳转换率为:"+(sum.toDouble/pageCount))
      }
    }
    sc.stop()
  }

  case class UserVisitAction(
        date: String, //用户点击行为的日期
        user_id: Long, //用户的 ID
        session_id: String, //Session 的 ID
        page_id: Long, //某个页面的 ID
        action_time: String, //动作的时间点
        search_keyword: String, //用户搜索的关键词
        click_category_id: Long, //某一个商品品类的 ID
        click_product_id: Long, //某一个商品的 ID
        order_category_ids: String, //一次订单中所有品类的 ID 集合
        order_product_ids: String, //一次订单中所有商品的 ID 集合
        pay_category_ids: String, //一次支付中所有品类的 ID 集合
        pay_product_ids: String, //一次支付中所有商品的 ID 集合
        city_id: Long //城市 id
                            )
}
